package com.dis.fan;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Flink CDC 2.2.1, 捕获MySQL数据表变化，并写入Oracle
 *
 * @author 闻武
 * @since 2020-05-31
 */
//todo: 目前有问题 20220601
public class CdcMySQL2Oracle {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String strSourceSql = "CREATE TABLE IF NOT EXISTS products_mys_cdc (\n" +
                "    id Int primary key,\n" +
                "    name String,\n" +
                "    description String\n" +
                "  ) with ( " +
                "    'connector' = 'mysql-cdc',\n" +
                "    'scan.startup.mode' = 'latest-offset',\n" +
                "    'hostname' = '192.168.123.58',\n" +
                "    'port' = '3306',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'database-name' = 'mydb',\n" +
                "    'table-name' = 'products',\n" +
                "    'debezium.log.mining.continuous.mine'='true',\n"+
                "    'debezium.log.mining.strategy'='online_catalog',\n" +
                "    'debezium.database.tablename.case.insensitive'='false',\n"+
                "    'scan.startup.mode' = 'initial')";

        String strSinkSql = " CREATE TABLE IF NOT EXISTS products_ora_sink (" +
                "    id INT,\n" +
                "    name STRING,\n" +
                "    description STRING,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                "  ) WITH (\n" +
                "    'connector' = 'jdbc'," +
                "    'url' = 'jdbc:oracle:thin:@192.168.123.58:1521:XE'," +
                "    'table-name' = 'products'," +
                "    'username' = 'flinkuser'," +
                "    'password' = 'flinkpw' " +
                " )";

        tableEnv.executeSql(strSourceSql);
        tableEnv.executeSql(strSinkSql);
        tableEnv.executeSql("insert into products_ora_sink select * from products_mys_cdc ");
    }
}